package com.ifind.async.executor;

import com.ifind.async.callback.DefaultGroupCallback;
import com.ifind.async.callback.IGroupCallback;
import com.ifind.async.wrapper.WorkerWrapper;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 类入口，可以根据自己情况调整core线程的数量
 *
 * @author: Carlos
 * @create: 2020/10/26 19:07
 */
public class Async {
    public static final ThreadPoolExecutor COMMON_POOL =
            new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2,  1024,
                                    15L, TimeUnit.SECONDS,
                                    new LinkedBlockingQueue<>(),
                                    (ThreadFactory) Thread::new);

    /**
     *
     * @param timeout
     * @param pool
     * @param workerWrappers
     * @return
     */
    public static boolean beginWork(long timeout, ThreadPoolExecutor pool, List<WorkerWrapper> workerWrappers) throws ExecutionException, InterruptedException {
        if (workerWrappers == null || workerWrappers.size() == 0) {
            return false;
        }
        Map<String, WorkerWrapper> forParamUseWrappers = new ConcurrentHashMap<>(16);
        CompletableFuture[] futures = new CompletableFuture[workerWrappers.size()];
        for (int i = 0; i < workerWrappers.size(); i++) {
            WorkerWrapper wrapper = workerWrappers.get(i);
            futures[i] = CompletableFuture.runAsync(() -> wrapper.work(pool, timeout, forParamUseWrappers), pool);
        }
        try {
            CompletableFuture.allOf(futures).get(timeout, TimeUnit.MILLISECONDS);
            return true;
        } catch (TimeoutException e) {
            Set<WorkerWrapper> set = new HashSet<>();
            totalWorkers(workerWrappers, set);
            for (WorkerWrapper workerWrapper : set) {
                workerWrapper.stopNow();
            }
            return false;
        }
    }

    /**
     * 如果想自定义线程池，请传pool。不自定义的话，就走默认的COMMON_POOL
     *
     * @param timeout
     * @param pool
     * @param workerWrappers
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static boolean beginWork(long timeout, ThreadPoolExecutor pool, WorkerWrapper... workerWrappers) throws ExecutionException, InterruptedException {
        if (workerWrappers == null || workerWrappers.length == 0) {
            return false;
        }
        List<WorkerWrapper> wrappers = Arrays.stream(workerWrappers).collect(Collectors.toList());
        return beginWork(timeout, pool, wrappers);
    }

    /**
     * 同步阻塞,直到所有都完成,或失败
     *
     * @param timeout
     * @param workerWrappers
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static boolean beginWork(long timeout, WorkerWrapper... workerWrappers) throws ExecutionException, InterruptedException {
        return beginWork(timeout, COMMON_POOL, workerWrappers);
    }

    /**
     * 异步执行,直到所有都完成,或失败后，发起回调
     *
     * @param timeout
     * @param groupCallback
     * @param workerWrappers
     */
    public static void beginWorkAsync(long timeout, IGroupCallback groupCallback, WorkerWrapper... workerWrappers) {
        if (groupCallback == null) {
            groupCallback = new DefaultGroupCallback();
        }
        IGroupCallback finalGroupCallback = groupCallback;
        COMMON_POOL.submit(() -> {
            try {
                boolean success = beginWork(timeout, COMMON_POOL, workerWrappers);
                if (success) {
                    finalGroupCallback.success(Arrays.asList(workerWrappers));
                } else {
                    finalGroupCallback.failure(Arrays.asList(workerWrappers), new TimeoutException());
                }
            } catch (ExecutionException |InterruptedException e) {
                e.printStackTrace();
                finalGroupCallback.failure(Arrays.asList(workerWrappers), e);
            }
        });
    }

    /**
     *
     * 总共多少个执行单元
     *
     * @param workerWrappers
     * @param set
     */
    @SuppressWarnings("unchecked")
    private static void totalWorkers(List<WorkerWrapper> workerWrappers, Set<WorkerWrapper> set) {
        set.addAll(workerWrappers);
        for (WorkerWrapper workerWrapper : workerWrappers) {
            if (workerWrapper.getNextWrappers() == null) {
                continue;
            }
            List<WorkerWrapper> wrappers = workerWrapper.getNextWrappers();
            totalWorkers(wrappers, set);
        }
    }

    public static void shutDown() {
        COMMON_POOL.shutdown();
    }

    public static String getThreadCount() {
        return "activeCount=" + COMMON_POOL.getActiveCount() +
                "  completedCount " + COMMON_POOL.getCompletedTaskCount() +
                "  largestCount " + COMMON_POOL.getLargestPoolSize();
    }
}
